package org.apache.hadoop.hdfs.server.namenode;

import java.io.IOException;
import java.io.OutputStream;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp.Writer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.IOUtils;

import com.google.common.base.Preconditions;

/**
 * EditsDoubleBuffer中包括两块缓存， 数据会先被写入到EditsDoubleBuffer的一块缓存中，
 * 而EditsDoubleBuffer的另一块缓存可能正在进行磁盘的同步操作（就是将缓存中的文件写入磁盘的操作） 。
 *
 * 每次双缓冲被刷新的时候（第2块buffer缓冲区被刷新到磁盘或者网络），两个buffer缓冲区会被交换一下
 *
 * EditsDoubleBuffer这样的设计会保证输出流进行磁盘同步操作的同时， 并不影响数据写入的功能。
 */
@InterfaceAudience.Private
public class EditsDoubleBuffer {
  // 里面放了2个缓冲区，每个缓冲区都是基于 DataOutputStream来实现的。
  // 其实说白了，就是将数据写入到内存中，这个是JDK IO 底层的API。



  private TxnBuffer bufCurrent; // 当前被写入的缓冲区
  private TxnBuffer bufReady;   // 正在进行flush的缓冲区
  private final int initBufferSize; // 初始化缓冲区大小   默认 512K

  public EditsDoubleBuffer(int defaultBufferSize) {
    // 传入进来的 size 就代表了一块缓冲区的大小
    initBufferSize = defaultBufferSize;
    bufCurrent = new TxnBuffer(initBufferSize);
    bufReady = new TxnBuffer(initBufferSize);
  }

  /**
   * 写入操作符至bufCurrent
   */
  public void writeOp(FSEditLogOp op) throws IOException {
    // 将数据写入到 bufCurrent 这块缓冲区里
    bufCurrent.writeOp(op);
  }

  /**
   * 写入事务至bufCurrent
   */
  public void writeRaw(byte[] bytes, int offset, int length) throws IOException {
    bufCurrent.write(bytes, offset, length);
  }
  
  public void close() throws IOException {
    Preconditions.checkNotNull(bufCurrent);
    Preconditions.checkNotNull(bufReady);

    int bufSize = bufCurrent.size();
    if (bufSize != 0) {
      throw new IOException("FSEditStream has " + bufSize + " bytes still to be flushed and cannot be closed.");
    }

    IOUtils.cleanup(null, bufCurrent, bufReady);
    bufCurrent = bufReady = null;
  }

  /**
   * 准备flush前， 首先要调用EditsDoubleBuffer.setReadyToFlush()方法交换两个缓冲区， 将正在写入的缓存改变为同步缓存， 然后才可以进行flush步操作。
   */
  public void setReadyToFlush() {
    // 确保之前的数据已经被flush完毕，调用isFlushed()方法判断bufReady的大小是否为0即可
    assert isFlushed() : "previous data not flushed yet";
    // 交换两个缓冲区
    TxnBuffer tmp = bufReady;
    bufReady = bufCurrent;
    bufCurrent = tmp;
  }
  
  /**
   * 完成了setReadyToFlush()调用之后， 输出流就可以调用flushTo()方法将flush缓存中的数据写入到磁盘文件中。
   */
  public void flushTo(OutputStream out) throws IOException {
    bufReady.writeTo(out);  // 将同步缓存中的数据写入到磁盘文件
    bufReady.reset();       // 将同步缓存中保存的数据清空
  }
  
  public boolean shouldForceSync() {
    // 非常关键的线索：必须当bufCurrent 缓冲区的数据 超过 initBufferSize的大小之后，才会认为一块缓冲区写满了，才会判断需要执行 fsync操作
    return bufCurrent.size() >= initBufferSize; // 默认大小，512KB
  }

  DataOutputBuffer getReadyBuf() {
    return bufReady;
  }
  
  DataOutputBuffer getCurrentBuf() {
    return bufCurrent;
  }

  public boolean isFlushed() {
    return bufReady.size() == 0;
  }

  public int countBufferedBytes() {
    return bufReady.size() + bufCurrent.size();
  }

  /**
   * @return the transaction ID of the first transaction ready to be flushed 
   */
  public long getFirstReadyTxId() {
    assert bufReady.firstTxId > 0;
    return bufReady.firstTxId;
  }

  /**
   * @return the number of transactions that are ready to be flushed
   */
  public int countReadyTxns() {
    return bufReady.numTxns;
  }

  /**
   * @return the number of bytes that are ready to be flushed
   */
  public int countReadyBytes() {
    return bufReady.size();
  }
  
  private static class TxnBuffer extends DataOutputBuffer {
    long firstTxId;
    int numTxns;
    private final Writer writer;
    
    public TxnBuffer(int initBufferSize) {
      super(initBufferSize);
      writer = new FSEditLogOp.Writer(this);
      reset();
    }

    public void writeOp(FSEditLogOp op) throws IOException {
      // 首次事务ID firstTxId被赋值为操作符的事务ID
      if (firstTxId == HdfsConstants.INVALID_TXID) {
        firstTxId = op.txid;
      } else {
        // 之后确保操作符的事务ID永远大于首次事务ID
        assert op.txid > firstTxId;
      }
      // 调用writer写入操作符
      writer.writeOp(op);
      numTxns++; // 事务数量numTxns累加
    }
    
    @Override
    public DataOutputBuffer reset() {
      super.reset();
      firstTxId = HdfsConstants.INVALID_TXID;
      numTxns = 0;
      return this;
    }
  }

}
